package cn.kgc.config;

import cn.kgc.service.MessageInfoService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

/**
 * RabbitMQ配置类，负责配置RabbitMQ的连接工厂、消息模板、消费者容器等核心组件。
 * <p>
 * 该配置类实现了以下功能：
 * 1. 配置RabbitMQ连接参数，建立与RabbitMQ服务器的连接
 * 2. 设置消息确认机制，确保消息可靠传递
 * 3. 配置消息转换器，支持JSON格式的消息序列化与反序列化
 * 4. 定义消费者容器工厂，配置并发消费参数
 * 5. 实现消息发送确认和返回机制的回调处理
 *
 * @author YC
 * 注解@EnableRabbit启用RabbitMQ注解驱动功能，使@RabbitListener等注解生效
 */
@Slf4j
@Configuration
@EnableRabbit
public class RabbitConfig {

    @Resource
    private MessageInfoService messageInfoService;

    /**
     * RabbitMQ服务器地址，格式为host:port，多个地址用逗号分隔
     */
    @Value("${mq.address}")
    private String address;

    /**
     * 连接RabbitMQ的用户名
     */
    @Value("${mq.username}")
    private String username;

    /**
     * 连接RabbitMQ的密码
     */
    @Value("${mq.password}")
    private String password;

    /**
     * RabbitMQ虚拟主机，用于实现多租户隔离
     */
    @Value("${mq.vhost}")
    private String vhost;

    /**
     * 当前服务名称，用于消息追踪和标识
     */
    @Value("${mq.serviceName}")
    private String serviceName;

    /**
     * 每个消费者预取的消息数量，用于流量控制
     */
    @Value("${mq.prefetchCount}")
    private int prefetchCount;

    /**
     * 消费者最小并发数
     */
    @Value("${mq.concurrentConsumers}")
    private int concurrentConsumers;

    /**
     * 消费者最大并发数
     */
    @Value("${mq.maxConcurrentConsumers}")
    private int maxConcurrentConsumers;


    /**
     * 配置RabbitTemplate，用于发送和接收消息的核心模板类。
     * <p>
     * 该模板配置了：
     * 1. JSON消息转换器，实现消息的序列化与反序列化
     * 2. 消息确认回调，处理消息是否成功到达交换机的通知
     * 3. 消息返回回调，处理消息到达交换机但未路由到队列的情况
     *
     * @param connectionFactory RabbitMQ连接工厂
     * @return 配置好的RabbitTemplate实例
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 设置JSON消息转换器，支持复杂对象的序列化与反序列化
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());

        // 配置消息确认回调：确认消息是否成功到达交换机
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack) {
                    log.debug("消息ID为{}的消息成功进入交换机，已从数据库中移除此消息", correlationData.getId());
                    // 消息成功发送到交换机后，更新数据库中消息状态
                    messageInfoService.messageDelete(correlationData.getId());
                } else {
                    log.warn("消息ID为{}的消息进入交换机失败", correlationData.getId());
                    // 此处可添加消息发送失败的处理逻辑，如重试或记录失败原因
                }
            }
        });

        // 启用强制返回机制：当消息无法路由到队列时，将消息返回给发送方
        rabbitTemplate.setMandatory(true);

        // 配置消息返回回调：处理消息到达交换机但未路由到队列的情况
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                // 从消息头中获取延迟时间，延迟消息不立即处理路由失败
//                long delayTime = message.getMessageProperties().getHeader("delayTime");
                long delayTime = (long) message.getMessageProperties().getHeaders().get("delayTime");
                if (delayTime <= 0L) {
                    log.warn("已进入交换机{}的消息ID为{}的消息未路由到队列，路由键：{}，原因：{}"
                            , exchange, message.getMessageProperties().getMessageId(), routingKey, replyText);

                    // 从消息头中获取发送次数，用于重试机制
//                    int sequence = message.getMessageProperties().getHeader("sequence");
                    int sequence = (int) message.getMessageProperties().getHeaders().get("sequence");
                    // 获取消息内容
                    String jsonStr = new String(message.getBody());
                    // 将路由失败的消息重新保存到数据库，等待重试
                    messageInfoService.messageSendReturn(exchange, routingKey, jsonStr, sequence, serviceName, delayTime);
                }
            }
        });
        return rabbitTemplate;
    }

    /**
     * 创建RabbitMQ连接工厂，配置连接参数并启用消息确认机制。
     * <p>
     * 连接工厂是创建RabbitMQ连接的核心组件，此处配置了：
     * 1. 服务器地址、虚拟主机、用户名和密码
     * 2. 启用发布者确认机制，确保消息成功到达交换机
     * 3. 启用发布者返回机制，处理无法路由到队列的消息
     *
     * @return 配置好的CachingConnectionFactory实例
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(address);
        connectionFactory.setVirtualHost(vhost);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        // 启用发布者返回机制，确认消息是否成功进入队列
        connectionFactory.setPublisherReturns(true);
        // 启用发布者确认机制，确认消息是否成功进入交换机，使用CORRELATED模式
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.createConnection();
        return connectionFactory;
    }

    /**
     * 创建RabbitAdmin实例，用于管理RabbitMQ中的交换机、队列和绑定关系。
     * <p>
     * 设置autoStartup为true，确保RabbitAdmin在应用启动时自动初始化。
     *
     * @param connectionFactory 连接工厂实例
     * @return 配置好的RabbitAdmin实例
     */
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        // 自动启动，确保admin功能生效
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    /**
     * 配置RabbitListener容器工厂，用于创建消息消费者容器。
     * <p>
     * 该工厂配置了：
     * 1. 手动确认模式，确保消息被正确处理后才确认消费
     * 2. 预取消息数量，控制消费者一次性获取的消息数量
     * 3. 并发消费参数，设置消费者的最小和最大并发数
     *
     * @param connectionFactory 连接工厂实例
     * @return 配置好的RabbitListenerContainerFactory实例
     */
    @Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        // 设置手动确认模式，需要在代码中显式调用channel.basicAck()确认消息
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 设置每个消费者预取的消息数量，防止消息堆积在单个消费者
        factory.setPrefetchCount(prefetchCount);
        // 设置消费者核心并发数
        factory.setConcurrentConsumers(concurrentConsumers);
        // 设置消费者最大并发数，支持动态扩缩容
        factory.setMaxConcurrentConsumers(maxConcurrentConsumers);
        // 关联连接工厂
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }

}
